package com.ksyun.campus.client;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.json.JSONUtil;
import com.ksyun.campus.client.constant.api.DataServerApi;
import com.ksyun.campus.client.constant.api.MetaServerApi;
import com.ksyun.campus.client.domain.*;
import com.ksyun.campus.client.util.FSHttpUtils;
import com.ksyun.campus.client.util.ZkUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class EFileSystem extends FileSystem {
    private String fileSystem = "default";

    private AtomicInteger atomicInteger = new AtomicInteger(0);

    public EFileSystem() {
        this.startZookeeper();
        createVolume();
    }

    public EFileSystem(String fileSystem) {
        //调用元数据服务 创建fileSystem ,  v
        String path = fileSystem.replaceAll("/", "");
        this.fileSystem = path;
        this.startZookeeper();
        createVolume();
    }


    public EFileSystem(String fileSystem, String zookeeperAddress) {
        //调用元数据服务 创建fileSystem ,  v
        String path = fileSystem.replaceAll("/", "");
        this.fileSystem = path;
        this.startZookeeper(zookeeperAddress);
        createVolume();
    }

    private void startZookeeper() {
        ZkUtil zkUtil = new ZkUtil();
        zkUtil.start();
    }

    private void startZookeeper(String adress) {
        ZkUtil zkUtil = new ZkUtil(adress);
        zkUtil.start();
    }


    /**
     * 创建磁盘
     */
    private void createVolume() {
        MetaServer metaMater = ZkUtil.getMetaMaster();
        log.info("等待获取元数据~");
        while (metaMater == null) {
            metaMater = ZkUtil.getMetaMaster();
        }
        log.info("成功获取~");
        String url = metaMater.getIp() + ":" + metaMater.getPort() + MetaServerApi.MKDIR;
        HashMap<String, String> param = new HashMap<>();
        param.put("type", "1");
        param.put("path", "");
        FSHttpUtils.get(url, param, fileSystem);
    }

    public FSInputStream open(String path) {
        //轮询选取元数据节点
        MetaServer metaServer = ZkUtil.getMetaMaster();
        //获取数据节点
        String url = metaServer.getIp() + ":" + metaServer.getPort() + MetaServerApi.OPEN;
        FSHttpUtils.HttpResult result = FSHttpUtils.get(url, path, fileSystem);
        if(result.code != 200){
            return new FSInputStream(new byte[0]);
        }
        List<ReplicaData> list = JSONUtil.toList(result.content, ReplicaData.class);
        for (int i = 0; i < list.size(); i++) {
            String openUrl = list.get(i).dsNode + DataServerApi.READ;
            byte[] read = FSHttpUtils.read(openUrl, path, fileSystem, 0, 0);
            if (read.length != 0) {
                return new FSInputStream(read);
            }
        }
        return new FSInputStream(new byte[0]);
    }

    /**
     * CAS获取一个idx
     *
     * @return 返回一个idx
     */
    private int getLoadBalanceAtomicIdx() {
        int current;
        int next;
        do {
            current = this.atomicInteger.get();
            next = current == Integer.MAX_VALUE ? 0 : current + 1;
        } while (!this.atomicInteger.compareAndSet(current, next));
        return next;
    }

    public FSOutputStream create(String path) {
        //先创建一个文件
        MetaServer metaMater = ZkUtil.getMetaMaster();
        assert metaMater != null;
        String url = metaMater.getIp() + ":" + metaMater.getPort() + MetaServerApi.CREATE;
        //创建文件元数据
        FSHttpUtils.get(url, path, fileSystem);
        //获取一个数据节点
        String pickUrl = metaMater.getIp() + ":" + metaMater.getPort() + MetaServerApi.PICK;
        FSHttpUtils.HttpResult result = FSHttpUtils.get(pickUrl, "", fileSystem);
        DataServer dataServer = JSONUtil.toBean(result.content, DataServer.class);
        return new FSOutputStream(dataServer.getIp() + ":" + dataServer.getPort(), fileSystem, path);
    }

    public boolean mkdir(String path) {
        //轮询选取元数据节点
        MetaServer metaServer = ZkUtil.getMetaMaster();
        String url = metaServer.getIp() + ":" + metaServer.getPort() + MetaServerApi.MKDIR;
        HashMap<String, String> map = new HashMap<>();
        map.put("path", path);
        map.put("type", "3");
        FSHttpUtils.HttpResult result = FSHttpUtils.get(url, map, fileSystem);
        if (result.code == 200) {
            return true;
        }
        return false;
    }

    public boolean delete(String path) {
        MetaServer metaMater = ZkUtil.getMetaMaster();
        String url = metaMater.getIp() + ":" + metaMater.getPort() + MetaServerApi.DELETE;
        FSHttpUtils.HttpResult result = FSHttpUtils.get(url, path, fileSystem);
        if (result.code == 200) {
            return true;
        }
        return false;
    }

    public StatInfo getFileStats(String path) {
        MetaServer metaServer = ZkUtil.getMetaMaster();
        String url = metaServer.getIp() + ":" + metaServer.getPort() + MetaServerApi.STATS;
        FSHttpUtils.HttpResult result = FSHttpUtils.get(url, path, fileSystem);
        if (result.code == 200) {
            StatInfo statInfo = JSONUtil.toBean(result.content, StatInfo.class);
            return statInfo;
        }
        return null;
    }

    public List<StatInfo> listFileStats(String path) {
        MetaServer metaServer = ZkUtil.getMetaMaster();
        String url = metaServer.getIp() + ":" + metaServer.getPort() + MetaServerApi.LIST_DIR;
        FSHttpUtils.HttpResult result = FSHttpUtils.get(url, path, fileSystem);
        if(result.code != 200){
            return null;
        }
        List<StatInfo> list = JSONUtil.toList(result.content, StatInfo.class);
        return list;
    }

    public ClusterInfo getClusterInfo() {
        ClusterInfo clusterInfo = new ClusterInfo();
        MetaServer metaMater = ZkUtil.getMetaMaster();
        ClusterInfo.MetaServerMsg master = new ClusterInfo.MetaServerMsg();
        if(metaMater == null){
            return null;
        }
        master.setHost(metaMater.getIp());
        master.setPort(metaMater.getPort());
        clusterInfo.setMasterMetaServer(master);

        MetaServer metaSlave = ZkUtil.getMetaSlave();
        ClusterInfo.MetaServerMsg slave = new ClusterInfo.MetaServerMsg();
        if (metaSlave != null) {
            slave.setHost(metaSlave.getIp());
            slave.setPort(metaSlave.getPort());
            clusterInfo.setSlaveMetaServer(slave);
        }
        String url = metaMater.getIp() + ":" + metaMater.getPort() + MetaServerApi.DATA_SERVER_LIST;
        FSHttpUtils.HttpResult result = FSHttpUtils.get(url, "", fileSystem);
        List<ClusterInfo.DataServerMsg> dataServerMsgs = new ArrayList<>();
        if (result.code == 200) {
            List<DataServer> list = JSONUtil.toList(result.content, DataServer.class);
            for (DataServer dataServer : list) {
                ClusterInfo.DataServerMsg dataServerMsg = new ClusterInfo.DataServerMsg();
                BeanUtil.copyProperties(dataServer, dataServerMsg, true);
                dataServerMsg.setHost(dataServer.getIp());
                dataServerMsgs.add(dataServerMsg);
            }
        }
        clusterInfo.setDataServer(dataServerMsgs);
        return clusterInfo;
    }

}
